package com.example.commom.utils;

import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Timer;
import java.util.TimerTask;

/**
 *  kafka 一些工具类
 */
@Slf4j
@Component
public class KafkaUtils {

    @Autowired
    private KafkaTemplate<String, byte[]> kafkaTemplate;


    private static final int MAX_RETRY_ATTEMPTS = 3;  // 最大重试次数
    private static final long INITIAL_RETRY_DELAY_MS = 500;  // 初始重试延迟时间（毫秒）
    private static final long MAX_RETRY_DELAY_MS = 5000;  // 最大重试延迟时间（毫秒）


    public void sendKafkaMessageWithRetry(String topic, String key, byte[] message) {

        sendWithRetry(topic, key, message, 0,
                INITIAL_RETRY_DELAY_MS, MAX_RETRY_ATTEMPTS, MAX_RETRY_DELAY_MS);
    }


    private void sendWithRetry(String topic,String key , byte[] message, int retryCount, long retryDelay, int maxRetryAttempts, long maxRetryDelay) {
        //              topic  key   message
        kafkaTemplate.send(topic, key, message)
            .addCallback(new ListenableFutureCallback<SendResult<String, byte[]>>() {
                @Override
                public void onSuccess(SendResult<String, byte[]> result) {
                    // 消息发送成功
                    log.info("消息发送成功: {}", result.getRecordMetadata());
                }

                @Override
                public void onFailure(@NotNull Throwable ex) {
                    // 消息发送失败
                    log.error("消息发送失败", ex);
                    if (shouldRetry(retryCount, maxRetryAttempts)) {
                        // 计算下一次的重试延迟时间
                        long nextRetryDelay = calculateRetryDelay(retryDelay, retryCount, maxRetryDelay);
                        // 执行重试逻辑
                        scheduleRetry(topic, key, message, retryCount + 1, nextRetryDelay, maxRetryAttempts, maxRetryDelay);
                    } else {
                        // 达到最大重试次数，进行其他错误处理
                        log.error("消息发送失败，超过最大重试次数");
                    }
                }
            });
    }

    private boolean shouldRetry(int retryCount, int maxRetryAttempts) {
        // 根据尝试次数和最大重试次数判断是否继续重试
        return retryCount < maxRetryAttempts;
    }

    private void scheduleRetry(String topic, String key, byte[] message, int retryCount, long retryDelay, int maxRetryAttempts, long maxRetryDelay) {
        // 定时任务，延迟一段时间后执行重试
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                sendWithRetry(topic, key, message, retryCount, retryDelay, maxRetryAttempts, maxRetryDelay);
            }
        }, retryDelay);
    }

    private long calculateRetryDelay(long previousDelay, int retryCount, long maxRetryDelay) {
        // 计算下一次的重试延迟时间（可根据实际需求进行调整）
        long nextDelay = previousDelay * (retryCount + 1);  // 每次重试延迟时间 * 重试次数
        nextDelay = Math.min(nextDelay, maxRetryDelay);  // 限制最大重试延迟时间
        return nextDelay;
    }


}
